package com.lagou.sparkgraphx

import org.apache.log4j.{Level, Logger}
import org.apache.spark.graphx.{Edge, Graph}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object GraphxTest {
  def main(args: Array[String]): Unit = {

    Logger.getLogger("org").setLevel(Level.ERROR)
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName).setMaster("local[*]")
    val sc = new SparkContext(conf)

    //初始化数据
    //定义顶点
    val vertexArray: Array[(Long, String)] = Array((1L,"SFO"),(2L, "ORD"),(3L, "DFW"))
    //定义边
    val edgeArray: Array[Edge[Int]] = Array(
      Edge(1L, 2L, 1800),
      Edge(2L, 3L, 800),
      Edge(3L, 1L, 1400)
    )

    // 构建vertexRDD和edgeRDD
    val vertexRDD: RDD[(Long, String)] = sc.makeRDD(vertexArray)
    val edgeRDD: RDD[Edge[Int]] = sc.makeRDD(edgeArray)

    // 构造图
    val graph: Graph[String, Int] = Graph(vertexRDD,edgeRDD)

    // 求所有的顶点
    println(s"所有顶点：")
    graph.vertices.foreach(println)

    // 求所有的边
    println(s"所有边：")
    graph.edges.foreach(println)

    // 求所有的triplets
    println(s"所有的triplets：")
    graph.triplets.foreach(println)

    //求顶点数
    val vertexCnt: Long = graph.vertices.count()
    println(s"顶点数为：${vertexCnt}")

    //求边数
    val edgeCnt: Long = graph.edges.count()
    println(s"边数为：${edgeCnt}")

    //求机场距离大于1000的有几个，有哪些
    val edgeLess: Long = graph.edges.filter(_.attr > 1000).count()
    println(s"机场距离大于1000的有：${edgeLess} 个")
    println("机场距离大于1000的有：" )
    graph.edges.filter(_.attr > 1000).foreach(println)

    // 按所有机场之间的距离排序（降序），输出结果
    println("所有机场之间距离(降序)：")
    graph.edges.sortBy(_.attr,false).collect.foreach(println)

  }
}
